package kim.artemis.rpc.core.protocol.server;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import kim.artemis.rpc.core.common.ArtemisDecoder;
import kim.artemis.rpc.core.common.ArtemisEncoder;
import kim.artemis.rpc.core.common.exception.ArtemisException;
import kim.artemis.rpc.core.model.ArtemisRpcRequest;
import kim.artemis.rpc.core.model.ArtemisRpcResponse;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.*;

/**
 * 服务端线程
 * @author Wq.
 */
@Slf4j
public class ArtemisRpcServerThread {

    private ArtemisRpcServer rpcServer;

    private Thread serverThread;

    /**
     * 启动线程
     * @param rpcServer
     */
    public void start (ArtemisRpcServer rpcServer) {
        this.rpcServer = rpcServer;
        serverThread = new ServerThread();
        serverThread.setDaemon(true);
        serverThread.start();
    }

    class ServerThread extends Thread  {

        @Override
        public void run() {
            // 创建服务线程池
            final ThreadPoolExecutor serverHandlerPool = serverThreadPool();
            EventLoopGroup bossGroup = new NioEventLoopGroup();
            EventLoopGroup workerGroup = new NioEventLoopGroup();
            try {
                ServerBootstrap bootstrap = new ServerBootstrap();
                bootstrap.group(bossGroup, workerGroup)
                        .channel(NioServerSocketChannel.class)
                        .childHandler(new ChannelInitializer<SocketChannel>() {
                            @Override
                            public void initChannel(SocketChannel channel) throws Exception {
                                channel.pipeline()
                                        // 反序列化
                                        .addLast(new ArtemisDecoder(ArtemisRpcRequest.class, rpcServer.getSerializer()))
                                        // 序列化
                                        .addLast(new ArtemisEncoder(ArtemisRpcResponse.class, rpcServer.getSerializer()))
                                        // 处理程序
                                        .addLast(new ArtemisRpcServerHandler(new ArtemisRpcServerInvoker(rpcServer.getAccessToken(), rpcServer.getServiceRepository()), serverHandlerPool));
                            }
                        })
                        .childOption(ChannelOption.TCP_NODELAY, true)
                        .childOption(ChannelOption.SO_KEEPALIVE, true);
                // Bind
                ChannelFuture future = bootstrap.bind(rpcServer.getPort()).sync();
                log.info(">>>>>>> artemis rpc, server started successfully, port: {}", rpcServer.getPort());
                future.channel().closeFuture().sync();
            } catch (Exception e) {
                if (e instanceof InterruptedException) {
                    log.info(">>>>>>> artemis rpc, remoting server stop.");
                } else {
                    log.error(">>>>>>> artemis rpc, remoting server error.", e);
                }
            } finally {
                try {
                    serverHandlerPool.shutdown();
                } catch (Exception e) {
                    log.error(">>>>>>> artemis rpc, close server handler thread error, {}", e.getMessage());
                }
                try {
                    workerGroup.shutdownGracefully();
                    bossGroup.shutdownGracefully();
                } catch (Exception e) {
                    log.error(">>>>>>> artemis rpc, close group error, {}", e.getMessage() );
                }
            }
        }
    }

    /**
     * 创建服务线程池
     * @return
     */
    public ThreadPoolExecutor serverThreadPool () {
        return new ThreadPoolExecutor(
                20,
                100,
                60L,
                TimeUnit.SECONDS,
                new LinkedBlockingQueue<Runnable>(1000),
                new ThreadFactory() {
                    @Override
                    public Thread newThread(Runnable r) {
                        return new Thread(r, "artemis rpc, serverHandlerPool-" + r.hashCode());
                    }
                },
                new RejectedExecutionHandler() {
                    @Override
                    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
                        throw new ArtemisException("artemis rpc, Thread pool is EXHAUSTED!");
                    }
                });
    }


    /***
     * 停止服务线程
     * @throws Exception
     */
    public void stop() throws Exception {
        if (null != serverThread && serverThread.isAlive()) {
            serverThread.interrupt();
        }
        log.info(">>>>>>> artemiss rpc, server destroy successfully.");
    }


}
